package com.laowang.flinkSQL;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.util.Arrays;
import java.util.Properties;

public class KafkaStreamSQL {
	public static void main(String[] args) throws Exception {

		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tEnv = TableEnvironment.getTableEnvironment(env);
		Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "localhost:9092");

        //books 的 kafka topic ，Flink-kafka
        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("books", new SimpleStringSchema(), properties);
        //从books这个topic的最开始位置进行消费
        consumer.setStartFromEarliest();

        DataStreamSource<String> topic = env.addSource(consumer);

        SingleOutputStreamOperator<String> kafkaSource = topic.map(new MapFunction<String, String>() {
            @Override
            public String map(String book) {
                return book;
            }
        });
        //注册内存表
        tEnv.registerDataStream("books", kafkaSource, "name");
        //SQL  select name, count(1) from books group by name;

        Table result = tEnv.sqlQuery("select name, count(1) from books group by name");

        //非常重要的知识点：回退更新  java：1 --》 java：2
        tEnv.toRetractStream(result, Row.class).print();

        //提交
        env.execute();
	}

}
